from abc import ABC, abstractmethod
from datetime import datetime
from enum import IntEnum
from typing import List, Dict, Union, Tuple
import numpy as np
import pandas as pd
from collections import deque
from easydict import EasyDict as edict
from importlib import resources
import functools
from ruamel import yaml
from matplotlib.dates import date2num
import aspectlib

# export_as_api by importing from rqalpha
# this is just to bypass pytype errors
logger = None
scheduler = None
is_suspended = None
order_target_value = None
order_target_percent = None
order_lots = None
subscribe = None
ORDER_STATUS = None


class TradeOp(IntEnum):
    HOLD = 0
    BUY = 1
    SELL = 2
    SKIP = 3
    OPEN = 4
    CLOSE = 5
    SKIP_BUY = 6
    SKIP_SELL = 7


# 每格大小为step，中值为mid
def grid_by_mid_step(mid: float, step: float, num_grid: int):
    assert num_grid % 2 == 0
    lower_bound = mid - step * (num_grid / 2 - 1)
    upper_bound = mid + step * (num_grid / 2)
    return grid_by_lh(lower_bound, upper_bound, num_grid)


# 每格大小为percent，中值为mid
def grid_by_mid_step_percent(mid: float, percent: float, num_grid: int):
    assert num_grid % 2 == 0
    grid: deque = deque()
    for mul in range(0, num_grid // 2):
        grid.append(mid * np.power((1 + percent), mul + 1))
        grid.appendleft(mid * np.power((1 - percent), mul))
    return np.array(list(grid))


def grid_by_lh(lower_bound: float, upper_bound: float, num_grid: int):
    grid = np.linspace(lower_bound, upper_bound, num_grid)
    return grid


@aspectlib.Aspect
def log_signal(*args, **kwargs):
    result = yield aspectlib.Proceed
    stself = args[0]
    stself.signals.append(result)
    yield aspectlib.Return(result)


def log_signal2(impl):
    def wrapper(*args):
        ret = impl(*args)
        stself = args[0]
        stself.signals.append(ret)
        return ret
    return wrapper


class TradeStrategy(ABC):
    def __init__(self, name: str):
        self.name = name
        self.signals = []

    @abstractmethod
    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        pass

    @abstractmethod
    def init(self, context):
        pass

    @abstractmethod
    def handle_bar(self, context, bar_dict):
        pass


class GridStrategy(TradeStrategy, ABC):
    def __init__(self, name: str, grid: List[float], order_book_id: str,
                 lots_per_grid: int):
        super().__init__(name)
        self.grid = grid
        self.order_book_id = order_book_id
        self.signal: pd.Series = None
        self.lots = 0
        self._lots_per_grid = lots_per_grid
        self.prev_dt = None
        self.suspended = False

    @property
    def grid(self):
        return self._grid

    @grid.setter
    def grid(self, value: List[float]):
        self._grid = value

    def grid_range(self,
                   grid_id: np.int64 = None) -> List[float]:
        if grid_id is None:
            return self.grid
        else:
            if grid_id.item() == 0:
                return [np.NINF, self.grid[0]]
            elif grid_id.item() == len(self.grid):
                return [self.grid[-1], np.Inf]
            else:
                return [self.grid[grid_id.item() - 1], self.grid[grid_id.item()]]

    @property
    def signal(self):
        return self._signal

    @signal.setter
    def signal(self, value: pd.Series):
        self._signal = value

    def __str__(self):
        return '网格大小:{0}，网格:{1}'.format(len(self.grid), self.grid)

    def init_state(self, signal: pd.Series):
        self.signal = signal
        self.lots = self.grid_lots(signal.grid_id)

    def grid_lots(self, grid_id: int) -> int:
        return (len(self.grid) -
                     grid_id) * self._lots_per_grid

    def grid_id(self, price: float) -> int:
        return np.searchsorted(self.grid, price)

    def make_signal(self, df: pd.DataFrame, now: datetime, order_book_id: str,
                    signal: TradeOp, grid_id: int, op: str, price: float):
        s = pd.Series({
            'order_book_id': order_book_id,
            'signal': signal,
            'grid_id': grid_id,
            'op': op,
            'price': price,
        })
        return df.join(s.to_frame().transpose().set_index('order_book_id'))

    def init(self, context):
        subscribe(self.order_book_id)
        logger.info('Backtesting %s on %s...' %
                    (self.name, self.order_book_id))

    def handle_bar(self, context, bar_dict):
        if self.prev_dt == context.now.date():
            if self.suspended:
                return

        self.prev_dt = context.now.date()
        self.suspended = is_suspended(self.order_book_id)
        if is_suspended(self.order_book_id):
            logger.error('%s suspended on: %s' %
                         (self.order_book_id, context.now.date()))
            return
        bar = bar_dict[self.order_book_id]
        data = {
            'order_book_id': bar.order_book_id,
            'datetime': context.now,
            'open': bar.open,
            'low': bar.low,
            'high': bar.high,
            'close': bar.close,
        }
        signal = self.generate_signal(pd.DataFrame([data]),
                                      context.now).iloc[0]
        if signal.signal in [
                TradeOp.OPEN, TradeOp.CLOSE, TradeOp.BUY, TradeOp.SELL
        ]:
            lots = self.grid_lots(signal.grid_id.item())
            order = order_lots(signal.order_book_id, lots - self.lots)
            #  order = order_target_percent(
            #  signal.order_book_id,
            #  1 - float(signal.grid_id) / len(self.grid.tolist()))
            if order is None:
                self.lots = lots
            elif order.status != ORDER_STATUS.FILLED:
                logger.error('Order error: %s' % order)


# 最基础的网格，只看当前区间决定买卖信号。
# 基础网格适用于中长线（例如按照日线进行），不适合短线（日内）或者根据分钟线数据进行交易
# 因为日内交易波动区间较小，可能在网格区间反复穿越。
class BasicGrid(GridStrategy):
    def __init__(self, grid: List[float], order_book_id: str, lots_per_grid=1):
        super().__init__('基础网格', grid, order_book_id, lots_per_grid)

    @log_signal
    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        assert len(df) == 1
        instrument = df.iloc[0]
        grid_id = self.grid_id(instrument.open)
        if self.signal is None:
            op = '建仓:%s' % grid_id
            s = TradeOp.OPEN
        elif self.signal.grid_id == grid_id:
            op = '格内震荡: %s' % grid_id
            s = TradeOp.HOLD
        elif grid_id < self.signal.grid_id:
            op = '加仓:%s->%s' % (self.signal.grid_id, grid_id)
            s = TradeOp.BUY
        else:
            op = '减仓:%s->%s' % (self.signal.grid_id, grid_id)
            s = TradeOp.SELL
        result = self.make_signal(df, now, instrument.name, s, grid_id, op,
                                  instrument.open)
        self.signal = result.iloc[0]
        return result


# 增强网格，如果两次买卖之间的差价小于阈值，不触发网格操作。
# 目的是避免反复穿越网格线造成频繁交易白交手续费
class EnhancedBasicGrid(GridStrategy):
    def __init__(self,
                 grid: List[float],
                 order_book_id: str,
                 threshold: float,
                 lots_per_grid=1):
        super().__init__('增强网格', grid, order_book_id, lots_per_grid)
        self._threshold = threshold

    @log_signal
    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        assert len(df) == 1
        instrument = df.iloc[0]
        grid_id = self.grid_id(instrument.open)
        if self.signal is None:
            op = '建仓:%s，%s' % (grid_id, self.grid_range(grid_id))
            s = TradeOp.OPEN
        elif self.signal.grid_id == grid_id:
            price_diff = instrument.open - self.signal.price
            gr = self.grid_range(grid_id)
            op = '格内震荡: {}，[{:.2f}, {:.2f}], 价差: {:.2f} ({:.2%})'.format(
                grid_id, gr[0], gr[1], price_diff,
                price_diff / self.signal.price)
            s = TradeOp.HOLD
        else:
            if self.signal.grid_id == 0:
                threshold = (self.grid[1] - self.grid[0])
            elif self.signal.grid_id == len(self.grid):
                threshold = (self.grid[-1] - self.grid[-2])
            else:
                threshold = self.grid[self.signal.grid_id -
                                      1] - self.grid[self.signal.grid_id - 2]
            threshold = threshold * self._threshold
            price_diff = instrument.open - self.signal.price
            if abs(price_diff) < threshold:
                op = '忽略信号:{0}->{1}，价格差abs({2:.2f} - {3:.2f}) = {4:.2f}({5:.2%})小于阈值{6:.2f}'.format(
                    self.signal.grid_id, grid_id, self.signal.price,
                    instrument.open, abs(price_diff),
                    price_diff / self.signal.price, threshold)
                s = TradeOp.SKIP
            else:
                if grid_id < self.signal.grid_id:
                    op = '加仓:{}->{}，{}, 价差: {:.2f} ({:.2%})'.format(
                        self.signal.grid_id, grid_id,
                        str(self.grid_range(grid_id)), price_diff,
                        price_diff / self.signal.price)
                    s = TradeOp.BUY
                else:
                    op = '减仓:{}->{}，{}, 价差: {:.2f} ({:.2%})'.format(
                        self.signal.grid_id, grid_id,
                        str(self.grid_range(grid_id)), price_diff,
                        price_diff / self.signal.price)
                    s = TradeOp.SELL

        result = self.make_signal(df, now, instrument.name, s, grid_id, op,
                                  instrument.open)
        if s in [TradeOp.OPEN, TradeOp.CLOSE, TradeOp.BUY, TradeOp.SELL]:
            self.signal = result.iloc[0]
        return result


# 动态网格根据买入卖出价格动态调整网格区间，每次的买入卖出价格会成为新的网格边界
# 目的是避免反复穿越网格线造成频繁交易白交手续费
class BasicDynamicGrid(GridStrategy):
    def __init__(self, grid: List[float], order_book_id: str, lots_per_grid=1):
        super().__init__('动态网格', grid, order_book_id, lots_per_grid)

    @log_signal
    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        assert len(df) == 1
        instrument = df.iloc[0]
        grid_id = self.grid_id(instrument.open)
        if self.signal is None:
            s = TradeOp.OPEN
            op = '建仓:%s，%s' % (grid_id, self.grid_range(grid_id))
        # 调整网格后，如果是买入，该价格的grid_id不变
        # 调整网格后，如果是卖出，该价格的grid_id减小了1格
        elif grid_id < self.signal.grid_id - (0 if self.signal.signal
                                              == TradeOp.BUY else 1):
            op = '加仓:%s->%s，%s' % (self.signal.grid_id, grid_id,
                                   self.grid_range(grid_id))
            s = TradeOp.BUY
        elif grid_id > self.signal.grid_id + (1 if self.signal.signal
                                              == TradeOp.BUY else 0):
            op = '减仓:%s->%s，%s' % (self.signal.grid_id, grid_id,
                                   self.grid_range(grid_id))
            s = TradeOp.SELL
        else:
            op = '格内震荡: %s，%s' % (grid_id, self.grid_range(grid_id))
            s = TradeOp.HOLD

        result = self.make_signal(df, now, instrument.name, s, grid_id, op,
                                  instrument.open)
        if s in [TradeOp.OPEN, TradeOp.BUY]:
            if grid_id < len(self.grid):
                diff = self.grid[grid_id] - instrument.open
                self.grid = self.grid - diff
            self.signal = result.iloc[0]

        if s in [TradeOp.CLOSE, TradeOp.SELL]:
            if grid_id > 0:
                diff = instrument.open - self.grid[grid_id - 1]
                self.grid = self.grid + diff
            self.signal = result.iloc[0]
        return result


# 拐点卖出条件单
class TurningPointSell(TradeStrategy):
    def __init__(self, threshold: float):
        self._base: pd.Series = None
        self._threshold = threshold

    def init(self, context):
        pass

    def handle_bar(self, context, bar_dict):
        pass

    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        assert len(df) == 1
        instrument = df.iloc[0]
        if self._base is None:
            self._base = instrument
            s = TradeOp.SKIP_SELL
            op = '等待减仓'
        elif instrument.open > self._base.open:
            self._base = instrument
            s = TradeOp.SKIP_SELL
            op = '新高'
        elif self._base.open - instrument.open < self._threshold:
            s = TradeOp.SKIP_SELL
            op = '高点回落'
        else:
            s = TradeOp.SELL
            op = '减仓'

        signal = pd.Series({
            'order_book_id': instrument.name,
            'signal': s,
            'op': op,
            'price': instrument.open,
            'tp': self._base.open,
        })
        return df.join(
            signal.to_frame().transpose().set_index('order_book_id'))


# 拐点买入条件单
class TurningPointBuy(TradeStrategy):
    def __init__(self, threshold: float):
        self._base: pd.Series = None
        self._threshold = threshold

    def init(self, context):
        pass

    def handle_bar(self, context, bar_dict):
        pass

    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        assert len(df) == 1
        instrument = df.iloc[0]
        if self._base is None:
            self._base = instrument
            s = TradeOp.SKIP_BUY
            op = '等待加仓'
        elif instrument.open < self._base.open:
            self._base = instrument
            s = TradeOp.SKIP_BUY
            op = '新低'
        elif instrument.open - self._base.open < self._threshold:
            s = TradeOp.SKIP_BUY
            op = '低点反弹'
        else:
            s = TradeOp.BUY
            op = '加仓'

        signal = pd.Series({
            'order_book_id': instrument.name,
            'signal': s,
            'op': op,
            'price': instrument.open,
            'tp': self._base.open,
        })
        return df.join(
            signal.to_frame().transpose().set_index('order_book_id'))


class TurningPointGridBase(GridStrategy):
    def __init__(self, name, grid: List[float], order_book_id: str,
                 bounce_turning_threshold: float,
                 drop_turning_threshold: float, lots_per_grid):
        super().__init__(name, grid, order_book_id, lots_per_grid)
        self._bounce_turning_threshold = bounce_turning_threshold
        self._drop_turning_threshold = drop_turning_threshold
        self._op = ''
        self._tp = 0.0

    def triggered_buy(self, price: float) -> bool:
        return False

    def triggered_sell(self, price: float) -> bool:
        return False

    def triggered_hold(self, price: float):
        pass

    def buy(self, price: float) -> int:
        return 0

    def sell(self, price: float) -> int:
        return 0

    @log_signal
    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        assert len(df) == 1
        instrument = df.iloc[0]
        if self.signal is None:
            s = TradeOp.OPEN
            grid_id = self.grid_id(instrument.open)
            op = '建仓:{0:.2f}'.format(instrument.open)
            price = instrument.open
        elif self.signal.signal == TradeOp.SKIP_SELL:
            if instrument.open > self._tp:
                s = TradeOp.SKIP_SELL
                grid_id = self.signal.grid_id
                gr = self.grid_range(grid_id)
                op = '新高:{0:.2f}, 前高:{1:.2f}, 涨幅:{2:.2%}'.format(
                    instrument.open, self._tp, instrument.open / gr[1] - 1)
                self._tp = instrument.open
                price = self.signal.price
            elif instrument.open < self._tp * (1 -
                                               self._drop_turning_threshold):
                grid_inc = self.sell(instrument.open)
                s = TradeOp.SELL
                grid_id = self.signal.grid_id + grid_inc
                op = '回落卖出:高点:{0:.2f}, 现价:{1:.2f}, 跌幅:{2:.2%}, 卖出{3}格'.format(
                    self._tp, instrument.open,
                    1 - (instrument.open / self._tp), grid_inc)
                price = instrument.open
            else:
                s = TradeOp.SKIP_SELL
                grid_id = self.signal.grid_id
                op = '高点回落:前高:{0:.2f}, 现价:{1:.2f}, 跌幅:{2:.2%}'.format(
                    self._tp, instrument.open,
                    1 - (instrument.open / self._tp))
                price = self.signal.price
        elif self.signal.signal == TradeOp.SKIP_BUY:
            if instrument.open < self._tp:
                s = TradeOp.SKIP_BUY
                grid_id = self.signal.grid_id
                gr = self.grid_range(grid_id)
                op = '新低:{0:.2f}, 前低:{1:.2f}, 跌幅:{2:.2%}'.format(
                    instrument.open, self._tp, 1 - instrument.open / gr[0])
                self._tp = instrument.open
                price = self.signal.price
            elif instrument.open > self._tp * (1 +
                                               self._bounce_turning_threshold):
                grid_inc = self.buy(instrument.open)
                s = TradeOp.BUY
                grid_id = self.signal.grid_id - grid_inc
                op = '反弹买入:低点:{0:.2f}, 现价:{1:.2f}, 涨幅:{2:.2%}, 买入{3}格'.format(
                    self._tp, instrument.open,
                    (instrument.open / self._tp) - 1, grid_inc)
                price = instrument.open
            else:
                s = TradeOp.SKIP_BUY
                grid_id = self.signal.grid_id
                op = '低点反弹:前低:{0:.2f}, 现价:{1:.2f}, 涨幅:{2:.2%}'.format(
                    self._tp, instrument.open,
                    (instrument.open / self._tp) - 1)
                price = self.signal.price
        elif self.triggered_sell(instrument.open):
            s = TradeOp.SKIP_SELL
            grid_id = self.signal.grid_id
            op = self._op
            price = self.signal.price
            self._tp = instrument.open
        elif self.triggered_buy(instrument.open):
            s = TradeOp.SKIP_BUY
            grid_id = self.signal.grid_id
            op = self._op
            price = self.signal.price
            self._tp = instrument.open
        else:
            self.triggered_hold(instrument.open)
            s = TradeOp.HOLD
            grid_id = self.signal.grid_id
            op = self._op
            price = self.signal.price

        result = self.make_signal(df, now, instrument.name, s, grid_id, op,
                                  price)
        self.signal = result.iloc[0]
        return result


class StaticTurningPointGrid(TurningPointGridBase):
    def __init__(self,
                 grid: List[float],
                 order_book_id: str,
                 bounce_turning_threshold: float,
                 drop_turning_threshold: float,
                 lots_per_grid=1):
        super().__init__('静态拐点网格', grid, order_book_id,
                         bounce_turning_threshold, drop_turning_threshold,
                         lots_per_grid)

    def triggered_buy(self, price: float) -> bool:
        grid_id = self.grid_id(price)
        gr = self.grid_range(grid_id)
        grid_size = gr[1] - gr[0]
        if grid_id < self.signal.grid_id:
            if self.signal.price - price >= grid_size:
                self._op = '触发买入:[{0:.2f}, {1:.2f}], 上次交易: {2:.2f}, 当前价格: {3:.2f}'.format(
                    gr[0], gr[1], self.signal.price, price)
                return True
            else:
                self._op = '网格变化:{0} -> {1}, 未触发买入, 上次交易: {2:.2f}, 当前价格: {3:.2f}'.format(
                    self.signal.grid_id, grid_id, self.signal.price, price)
                return False
        else:
            gr = self.grid_range(self.signal.grid_id)
            self._op = '格内震荡:[{0:.2f}, {1:.2f}]'.format(gr[0], gr[1])
            return False

    def triggered_sell(self, price: float) -> bool:
        grid_id = self.grid_id(price)
        gr = self.grid_range(grid_id)
        grid_size = gr[1] - gr[0]
        if grid_id > self.signal.grid_id:
            if price - self.signal.price >= grid_size:
                self._op = '触发卖出:[{0:.2f}, {1:.2f}], 上次交易: {2:.2f}, 当前价格: {3:.2f}'.format(
                    gr[0], gr[1], self.signal.price, price)
                return True
            else:
                self._op = '网格变化:{0} -> {1}, 未触发卖出, 上次交易: {2:.2f}, 当前价格: {3:.2f}'.format(
                    self.signal.grid_id, grid_id, self.signal.price, price)
                return False
        else:
            gr = self.grid_range(self.signal.grid_id)
            self._op = '格内震荡:[{0:.2f}, {1:.2f}]'.format(gr[0], gr[1])
            return False

    def triggered_hold(self, price: float):
        # Message set by triggered_buy / triggered_sell
        pass

    def buy(self, price: float) -> int:
        grid_id = self.grid_id(price)
        grid_inc = self.signal.grid_id - grid_id
        if grid_inc == 0:
            grid_inc = 1
        return grid_inc

    def sell(self, price: float) -> int:
        grid_id = self.grid_id(price)
        grid_inc = grid_id - self.signal.grid_id
        if grid_inc == 0:
            grid_inc = 1
        return grid_inc

    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        return super().generate_signal(df, now)


# 拐点网格
# 更适用与根据分钟线进行交易
class DynamicTurningPointGrid(TurningPointGridBase):
    def __init__(self,
                 grid: List[float],
                 order_book_id: str,
                 percent: float,
                 bounce_turning_threshold: float,
                 drop_turning_threshold: float,
                 lots_per_grid=1):
        super().__init__('动态拐点网格', grid, order_book_id,
                         bounce_turning_threshold, drop_turning_threshold,
                         lots_per_grid)
        self._pct = percent
        self._base: float = None

    def init_state(self, signal: pd.Series):
        super().init_state(signal)
        if signal is not None:
            self._base = signal.price
        else:
            self._base = None

    def triggered_buy(self, price: float) -> bool:
        if price < self._base * (1 - self._pct):
            self._op = '触发买入:基准:{0:.2f}, 现价:{1:.2f}, 跌幅:{2:.2%}'.format(
                self._base, price, self._base / price - 1)
            return True
        return False

    def triggered_sell(self, price: float) -> bool:
        if price > self._base * (1 + self._pct):
            self._op = '触发卖出:基准:{0:.2f}, 现价:{1:.2f}, 涨幅:{2:.2%}'.format(
                self._base, price, price / self._base - 1)
            return True
        return False

    def triggered_hold(self, price: float):
        self._op = '格内震荡:基准:{0:.2f}, 网格:[{1:.2f}, {2:.2f}], 现价:{3:.2f}, 距离基准:{4:.2%}'.format(
            self._base, self._base * (1 - self._pct),
            self._base * (1 + self._pct), price, price / self._base -
            1 if price >= self._base else price / self._base - 1)

    def buy(self, price: float) -> int:
        grid_inc = int((self._base - price) / price / self._pct)
        if grid_inc == 0:
            grid_inc = 1
        self._base = price
        return grid_inc

    def sell(self, price: float) -> int:
        grid_inc = int((self._tp - self._base) / self._base / self._pct)
        if grid_inc == 0:
            grid_inc = 1
        self._base = price
        return grid_inc

    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        signal = super().generate_signal(df, now)
        if self.signal.signal == TradeOp.OPEN:
            self._base = self.signal.price
        signal['base'] = self._base
        return signal


# 转债轮动策略
class ConbondRotateStrategy(TradeStrategy, ABC):
    def __init__(self, name: str, bars: pd.DataFrame,
                 instruments: pd.DataFrame):
        with resources.path('quant', 'conbond.yml') as f:
            with open(f, mode='r') as fp:
                strategies = {
                    doc['name']: doc
                    for doc in yaml.safe_load_all(fp)
                }
        self.cfg = edict(strategies[name])
        super().__init__(self.cfg.description)
        self._bars = bars
        self._instruments = instruments

    def eval_fn(self, now: datetime, expr: str, bond: pd.Series):
        values = bond.to_dict()
        values['now'] = date2num(now)
        stmt = expr.format(**values)
        try:
            return eval(stmt)
        except Exception as e:
            logger.error(stmt)
            logger.error(values)
            raise e

    def filter(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        if not 'filters' in self.cfg:
            return df

        def filter_fn(bond, conditions=self.cfg.filters, eval_fn=self.eval_fn):
            for cond in conditions:
                for reason, expr in cond.items():
                    if eval_fn(now, expr, bond):
                        return reason
            return ''

        result = df.copy()
        result['filtered'] = result.apply(filter_fn, axis=1)
        return result

    def rank(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        if not 'rankings' in self.cfg:
            return df

        idx_names = df.index.names
        result = df.reset_index()
        for ranking in self.cfg.rankings:
            for col, expr in ranking.items():
                name = 'rank_by_%s' % col
                result[name] = result.apply(functools.partial(
                    self.eval_fn, now, expr),
                                            axis=1)
                r = result.sort_values(name).reset_index(drop=True)
                result[col] = r.index.to_series()
        result.set_index(idx_names, inplace=True)
        return result

    def score(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        result = df.copy()
        result['__score__'] = result.apply(functools.partial(
            self.eval_fn, now, self.cfg.score),
                                           axis=1)
        return result

    @log_signal
    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        return self.explain(df, now)[0]

    def explain(
            self, df: pd.DataFrame,
            now: datetime) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
        filtered = self.filter(df, now)
        ranked = self.rank(filtered[(filtered.filtered == '')], now)
        scored = self.score(ranked, now)
        return scored.nsmallest(
            self.cfg.top,
            '__score__'), filtered[~(filtered.filtered == '')], ranked

    def init(self, context):
        logger.info('Backtesting %s...' % self.name)
        scheduler.run_weekly(self.rebalance, tradingday=1)

    def handle_bar(self, context, bar_dict):
        pass

    def rebalance(self, context, bar_dict):
        dt = context.now
        data = self._bars.loc(axis=0)[pd.IndexSlice[dt, :]]
        data = data.reset_index().set_index('order_book_id').join(
            self._instruments[['bond_type', 'symbol',
                               'maturity_day']]).reset_index().set_index(
                                   ['datetime', 'order_book_id'])
        data = data[data.bond_type == 'cb']
        signal = self.generate_signal(data, dt)
        if context is None:
            logger.info(signal)
        else:
            candidates = set(signal.index.get_level_values(1).tolist())

            positions = set()
            for p in context.portfolio.stock_account.get_positions():
                positions.add(p.order_book_id)

            suspended = set()
            # close
            for target in (positions - candidates):
                if is_suspended(target):
                    suspended.add(target)
                    logger.error('Not closing %s as it is suspended.' % target)
                else:
                    order = order_target_percent(target, 0)
                    if order is not None and order.status != ORDER_STATUS.FILLED:
                        logger.error('Order error: %s' % order)

            # adjust
            for order_book_id in candidates:
                order = order_target_value(
                    order_book_id, context.portfolio.cash / len(candidates))
                # order can be None if the quantity is 0
                if order is not None and order.status != ORDER_STATUS.FILLED:
                    logger.error('Order error: %s' % order)


class Benchmark(TradeStrategy, ABC):
    def __init__(self, name, order_book_id):
        super().__init__(name)
        self.order_book_id = order_book_id

    @log_signal
    def generate_signal(self, df: pd.DataFrame, now: datetime) -> pd.DataFrame:
        pass

    def init(self, context):
        logger.info('Backtesting benchmark %s(%s)...' %
                    (self.name, self.order_book_id))
        context.fired = False

    def handle_bar(self, context, bar_dict):
        if context.fired:
            return
        else:
            context.fired = True
            order = order_target_percent(self.order_book_id, 1)
            if order is not None and order.status != ORDER_STATUS.FILLED:
                logger.error('Order error: %s' % order)
